package rabbitmq

import (
	"context"
	"portal_mq/global"
	"portal_mq/models/db"

	"github.com/FISCO-BCOS/go-sdk/core/types"
	"github.com/ethereum/go-ethereum/common"
	amqp "github.com/rabbitmq/amqp091-go"
	"go.uber.org/zap"
)

type RabbitMQ struct {
	conn         *amqp.Connection
	channel      *amqp.Channel
	QueueName    string
	ExchangeName string
	BindingKey   string
}

type MQManager struct {
	TxConsumer1 *RabbitMQ // same queue name
	TxConsumer2 *RabbitMQ // same queue name
	closeCh     chan (bool)
}

func InitRabbitMQ(isDebug bool) {

	portalTxConsumer1 := NewRabbitMQTopicConsumer(&RabbitOptions{
		QueueName:    global.ServiceConfig.TxQueue.Queue, // same queue
		ExchangeName: global.ServiceConfig.TxQueue.Exchange,
		BindingKey:   global.ServiceConfig.TxQueue.BindingKey,
	})
	portalTxConsumer2 := NewRabbitMQTopicConsumer(&RabbitOptions{
		QueueName:    global.ServiceConfig.TxQueue.Queue, // same queue
		ExchangeName: global.ServiceConfig.TxQueue.Exchange,
		BindingKey:   global.ServiceConfig.TxQueue.BindingKey,
	})
	global.Manager = &MQManager{
		TxConsumer1: portalTxConsumer1,
		TxConsumer2: portalTxConsumer2,
		closeCh:     make(chan bool),
	}
	global.Manager.Start()
	zap.L().Info("rabbitmq initialized")
}

func (m *MQManager) Start() {

	go func(m *MQManager) {
		msg1 := m.TxConsumer1.Consume()
		msg2 := m.TxConsumer2.Consume()

		for {
			select {
			case msg := <-msg1:
				// TODO: goroutine pool
				// send message
				zap.L().Debug("got msg from consumer1", zap.String("msg", string(msg.Body)))
				hexStr := string(msg.Body)
				go func(m *MQManager, hexStr string) {
					defer func() {
						if err := recover(); err != nil {
							zap.L().Error("运行时错误", zap.Any("err", err))
						}
					}()
					txHash := common.HexToHash(hexStr)
					m.HandleTxHashMessage(txHash)
				}(m, hexStr)

			case msg := <-msg2:
				// save in database
				zap.L().Debug("got msg from consumer2", zap.String("msg", string(msg.Body)))
				hexStr := string(msg.Body)
				go func(m *MQManager, hexStr string) {
					defer func() {
						if err := recover(); err != nil {
							zap.L().Error("运行时错误", zap.Any("err", err))
						}
					}()
					txHash := common.HexToHash(hexStr)
					m.HandleTxHashMessage(txHash)
				}(m, hexStr)

			case <-m.closeCh:
				zap.L().Debug("close goroutine1")
				m.TxConsumer1.Close()
				m.TxConsumer2.Close()
				return
			}
		}
	}(m)

}

func (m *MQManager) Close() {
	m.closeCh <- true
}

func (m *MQManager) HandleTxHashMessage(txHash common.Hash) {
	// get tx
	receipt, err := global.Client.GetTransactionReceipt(context.Background(), txHash)
	if err != nil {
		zap.L().Error("get transaction receipt failed", zap.Error(err))
		return
	}
	if receipt.Status == types.Success {
		// 更新状态为成功
		// 0-created,1-success,2-failed,3-unknown
		err = db.UpdatePortalRecordStatus(txHash.Hex(), 1)
		if err != nil {
			zap.L().Error("update portal record status failed", zap.Error(err))
			return
		}
	} else if receipt.Status == types.Unknown {
		err = db.UpdatePortalRecordStatus(txHash.Hex(), 3)
		if err != nil {
			zap.L().Error("update portal record status failed", zap.Error(err))
			return
		}
	} else {
		err = db.UpdatePortalRecordStatus(txHash.Hex(), 2)
		if err != nil {
			zap.L().Error("update portal record status failed", zap.Error(err))
			return
		}
	}
}
